package com.sali.jsondecorator;

import com.sali.common.KafkaUtils;
import com.sali.pojo.KafkaEntity;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class JsonConsumer extends KafkaUtils {

    public static void consumeData(KafkaEntity kafkaEntity) {
        checkParams(kafkaEntity);

        Properties kafkaProperties = getKafkaProperties(kafkaEntity.getBrokerIp());
        kafkaProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        kafkaProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        kafkaProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaEntity.getConsumeOffset());
        kafkaProperties.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaEntity.getGroupId());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafkaProperties);
        consumer.subscribe(Collections.singleton(kafkaEntity.getKafkaTopic()));

        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("接收到的数据为：" + record.value());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        final String brokerIp = "192.168.80.103:6667";
        final String kafkaTopic = "zh_json_topic001";
        KafkaEntity kafkaEntity = new KafkaEntity();
        kafkaEntity.setBrokerIp(brokerIp);
        kafkaEntity.setKafkaTopic(kafkaTopic);

        while (true) {
            consumeData(kafkaEntity);
            Thread.sleep(2 * 1000);
        }
    }
}
